#ifndef A8_SOCKET_H
#define A8_SOCKET_H

#include <string>
#include <errno.h>
#include <stdio.h>
#include <a8/delegate.hpp>
#include <a8/thread.hpp>
#include <a8/sync.hpp>
#include <a8/list.hpp>
#include <a8/sysutils.hpp>

#ifdef WIN32   
#include <WinSock2.h>
typedef int socklen_t;
typedef int ssize_t;
#endif   
#if defined(__linux__) || defined(__FREEBSD__) || defined(__APPLE__)
typedef int SOCKET;   
#define SOCKET_ERROR (-1)   
#define INVALID_SOCKET (-1)
#define closesocket(h) close(h)
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <strings.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <unistd.h>
#endif   
#if defined(__FREEBSD__) || defined(__APPLE__)
#include <sys/event.h>
#endif
#if defined(__linux__)
#include <sys/epoll.h>
#endif

#define INVALID_SOCKET_HANDLE (0)

namespace a8
{
	const unsigned short MAX_RECV_BUFFERSIZE = 1024 * 8;

	struct SendQueueNode
	{
		char* mBuff;
		unsigned int mBuffLen;
		unsigned int mSentBytes;
		SendQueueNode *mNext;
	};

#ifdef WIN32
	const unsigned short MAX_IOCP_BUF_SIZE = MAX_RECV_BUFFERSIZE;
	enum IOCP_OPCODE
	{
		IOCP_RECV,
		IOCP_SEND
	};

	struct IocpOverLapped
	{
		unsigned short mSocketHandle;
		WSABUF mWsaBuf;
		OVERLAPPED mOverLapped;
		IOCP_OPCODE mOpCode;
		char mBuffer[MAX_IOCP_BUF_SIZE];
	};

	const unsigned int DISCONNECT_FLAG  = 0xFFFFFFFE;
#endif

	class UdpClient
	{
	public:
		std::string mRemoteAddress;
		int mRemotePort;

		UdpClient():mRemotePort(0), mSocket(INVALID_SOCKET), mTopNode(NULL), mBotNode(NULL),
					mSenderThread(NULL)
		{
		}

		virtual ~UdpClient()
		{
			Close();
		}

		void Open()
		{
			if(!IsActive()){
				Close();
				mSocket = socket(AF_INET, SOCK_DGRAM, 0);
				if(INVALID_SOCKET == mSocket){
					//					OnError(this, errno);
					return;
				}
				memset(&mAddr, sizeof(mAddr), 0);
				mAddr.sin_family = AF_INET;
				mAddr.sin_port = htons(mRemotePort);
				mAddr.sin_addr.s_addr = inet_addr(mRemoteAddress.c_str());
				mSenderThread = new a8::Runnable();
				mSenderThread->OnExecute += std::make_pair(this, &UdpClient::SenderThreadProc);
				mSenderThread->Start();
			}
		}

		void Close()
		{
			if(IsActive()){
				closesocket(mSocket);				
				mSocket = INVALID_SOCKET;
				if(mSenderThread){
					delete mSenderThread;
					mSenderThread = NULL;
				}
			}
		}

		bool IsActive()
		{
			return mSocket != INVALID_SOCKET;
		}

		void SendBuff(const char* buff, unsigned int bufflen)
		{	
			if (bufflen > 0){
				SendQueueNode* p = (SendQueueNode*)malloc(sizeof(SendQueueNode));
				memset(p, 0, sizeof(SendQueueNode));
				p->mBuff = (char*)malloc(bufflen);
				memmove(p->mBuff, buff, bufflen);
				p->mBuffLen = bufflen;			
				mSendBufferMutex.Lock();
				if (mBotNode){
					mBotNode->mNext = p;
					mBotNode = p;
				}else{
					mTopNode = p;
					mBotNode = p;
				}
				mSendBufferMutex.UnLock();
			}
		}

	private:
		volatile SOCKET mSocket;
		a8::sync::Cond mSendBufferCond;
		a8::sync::Mutex mSendBufferMutex;
		SendQueueNode *mTopNode;
		SendQueueNode *mBotNode;
		a8::Runnable *mSenderThread;
		struct sockaddr_in mAddr;

		void SenderThreadProc(a8::Runnable* thread)
		{
			SendQueueNode *worknode = NULL, *currnode = NULL;			
			while(!thread->IsShutdown()){
				if(!worknode && mTopNode){
					mSendBufferMutex.Lock();
					worknode = mTopNode;
					mTopNode = NULL;
					mBotNode = NULL;
					mSendBufferMutex.UnLock();
				}

				while(worknode && !thread->IsShutdown()){
					currnode = worknode;
					int len = ::sendto(mSocket, currnode->mBuff, currnode->mBuffLen, 0, (struct sockaddr*)&mAddr, sizeof(mAddr));
					if(len > 0){
					}
					worknode = worknode->mNext;
					free(currnode->mBuff);
					free(currnode);
				}
				mSendBufferCond.Wait(10);
			}
			while(worknode){
				currnode = worknode;
				worknode = worknode->mNext;
				free(currnode->mBuff);
				free(currnode);
			}
		}

	};

	class UdpListener
	{
	public:
		a8::Delegate<void (a8::UdpListener*, int errorId)> OnError;
		std::string mBindAddress;
		unsigned short mBindPort;

		UdpListener():
			mListenSocket(INVALID_SOCKET), mBindPort(0), mWorkerThread(NULL)
		{
		}

		virtual ~UdpListener()
		{
		}

		virtual void OnSocketRead(char*, unsigned int){}
		
		void Open()
		{
			if(!isActive())
				setActive(true);
		}
	 
		void Close()
		{
			if(isActive())
				setActive(false);
		}
	 
		bool isActive() 
		{
			return mListenSocket != INVALID_SOCKET;
		}

	private:
		SOCKET mListenSocket;
		a8::Runnable* mWorkerThread;

		void setActive(bool active) 
		{ 
			if(active){
				if(isActive()) 
					return;
				if(!mWorkerThread){
					mWorkerThread = new a8::Runnable();
					mWorkerThread->OnExecute += std::make_pair(this, &UdpListener::WorkerThreadProc);
					mWorkerThread->Start();
				}
			}else{
				if(isActive())
					ActiveStop();		
			}
		}

		bool ActiveStart()
		{
			mListenSocket = socket(AF_INET, SOCK_DGRAM, 0);
			if(INVALID_SOCKET == mListenSocket){
				OnError(this, errno);
				return false;
			}
			// TIME_WAIT - argh
			int on = 1;
			if (setsockopt(mListenSocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on)) == -1 ){
				//return false;
			}
			sockaddr_in sa;
			memset(&sa, 0, sizeof(sa));
			sa.sin_family = AF_INET;
			sa.sin_addr.s_addr = inet_addr(mBindAddress.c_str());
			sa.sin_port = htons(mBindPort);
			if(::bind(mListenSocket, (sockaddr*)&sa, sizeof(sa)) < 0){
				OnError(this, errno);
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;
				return false;
			}
			return true;
		}

		void ActiveStop()
		{
			if(mListenSocket != INVALID_SOCKET){
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;								
			}
			if (mWorkerThread){
				delete mWorkerThread;
				mWorkerThread = NULL;
			}
		}

		void WorkerThreadProc(a8::Runnable* thread)
		{
			if(!ActiveStart())
				return;
			struct sockaddr_in destAddr;
			socklen_t addrLen;
			char recvBuf[MAX_RECV_BUFFERSIZE];						
			while(!thread->IsShutdown()){
				int ret = ::recvfrom(mListenSocket, recvBuf, (size_t)sizeof(recvBuf), 0, (struct sockaddr*)&destAddr, (socklen_t*)&addrLen);
				if(ret > 0){
					OnSocketRead(recvBuf, ret);
				}
			}
		}

	};

	class TcpClient
	{
	public:
		a8::Delegate<void (TcpClient*, int)> OnError;
		a8::Delegate<void (TcpClient*)> OnConnect;
		a8::Delegate<void (TcpClient*)> OnDisConnect;
		a8::Delegate<void (TcpClient*, char*, int)> OnSocketRead;
		std::string mRemoteAddress;
		int mRemotePort;

		TcpClient():mSocket(INVALID_SOCKET), mConnected(false), mRemoteAddress(""), mRemotePort(0), 
			mWorkerThread(NULL), mTopNode(NULL), mBotNode(NULL)
		{
		}

		virtual ~TcpClient()
		{
			Close();
		}

		void Open()
		{
			if(!IsActive()){
				SetActive(true);
			}
		}

		void Close()
		{
			if(IsActive()){
				SetActive(false);
			}
		}

		bool IsActive() 
		{ 
			return mSocket != INVALID_SOCKET; 
		}    

		bool Connected() 
		{ 
			return mConnected;
		}

		void SendBuff(const char* buff, unsigned int bufflen)
		{	
			if (bufflen > 0){
				SendQueueNode* p = (SendQueueNode*)malloc(sizeof(SendQueueNode));
				memset(p, 0, sizeof(SendQueueNode));
				p->mBuff = (char*)malloc(bufflen);
				memmove(p->mBuff, buff, bufflen);
				p->mBuffLen = bufflen;			
				mSendBufferMutex.Lock();
				if (mBotNode){
					mBotNode->mNext = p;
					mBotNode = p;
				}else{
					mTopNode = p;
					mBotNode = p;
				}
				mSendBufferMutex.UnLock();
			}
		}

	private:
		volatile SOCKET mSocket;
		bool mConnected;
		a8::Runnable* mWorkerThread;
		a8::sync::Cond mSendBufferCond;
		a8::sync::Mutex mSendBufferMutex;
		SendQueueNode *mTopNode;
		SendQueueNode *mBotNode;

		void SetActive(bool active) 
		{	 
			if(active){
				if(IsActive())
					return;
				if(!mWorkerThread){
					ActiveStop();
				}
				mWorkerThread = new a8::Runnable();
				mWorkerThread->OnExecute += std::make_pair(this, &TcpClient::WorkerThreadProc);
				mWorkerThread->Start();
			}else{
				if(mWorkerThread){
					ActiveStop();
				}
			}
		}

		bool ActiveStart()
		{
			mSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
			if(INVALID_SOCKET == mSocket){
				OnError(this, errno);
				return false;
			}
			sockaddr_in sa;
			memset(&sa, 0, sizeof(sa));
			sa.sin_family = AF_INET;
			sa.sin_addr.s_addr = inet_addr(mRemoteAddress.c_str());
			sa.sin_port = htons(mRemotePort);
			if(connect(mSocket, (sockaddr*)&sa, sizeof(sa)) < 0){
				OnError(this, errno);
				closesocket(mSocket);
				mSocket = INVALID_SOCKET;
				return false;
			}
			mConnected = true;
			OnConnect(this);
			return true;
		}

		void ActiveStop()
		{
			mConnected = false;
			if(mSocket != INVALID_SOCKET){
				shutdown(mSocket, 2);
				closesocket(mSocket);				
			}
			if(mWorkerThread){
				delete mWorkerThread;
				mWorkerThread = NULL;
			}
			mSocket = INVALID_SOCKET;
		}

		void WorkerThreadProc(a8::Runnable* thread)
		{
			if(!ActiveStart())
				return;

			a8::Runnable *senderthread = new a8::Runnable();
			senderthread->OnExecute += std::make_pair(this, &TcpClient::SenderThreadProc);
			senderthread->Start();
			char recvBuf[MAX_RECV_BUFFERSIZE];						
			while(!thread->IsShutdown()){
				int ret = ::recv(mSocket, recvBuf, MAX_RECV_BUFFERSIZE, 0);
				if(ret < 0){
					mConnected = false;
					OnDisConnect(this);
					mWorkerThread->Stop();
					break;
				}else if(ret == 0){
					mConnected = false;
					OnDisConnect(this);
					mWorkerThread->Stop();
					break;
				}else{
					OnSocketRead(this, recvBuf, ret);
				}
			}
			delete senderthread;
			mSocket = INVALID_SOCKET;
		}

		void SenderThreadProc(a8::Runnable* thread)
		{
			SendQueueNode *worknode = NULL, *currnode = NULL;			
			while(!thread->IsShutdown()){
				if(!worknode && mTopNode){
					mSendBufferMutex.Lock();
					worknode = mTopNode;
					mTopNode = NULL;
					mBotNode = NULL;
					mSendBufferMutex.UnLock();
				}

				while(worknode && !thread->IsShutdown()){
					currnode = worknode;
					while(currnode->mSentBytes < currnode->mBuffLen && !thread->IsShutdown()){
						int len = ::send(mSocket, currnode->mBuff + currnode->mSentBytes, 
										 currnode->mBuffLen - currnode->mSentBytes, 0);
						if(len > 0){
							currnode->mSentBytes += len;
						}else{
							break;
						}
					}
					//send
					if(currnode->mSentBytes >= currnode->mBuffLen){
						worknode = worknode->mNext;
						free(currnode->mBuff);
						free(currnode);
					}
				}
				mSendBufferCond.Wait(10);
			}
			while(worknode){
				currnode = worknode;
				worknode = worknode->mNext;
				free(currnode->mBuff);
				free(currnode);
			}
		}

	};

	class ClientSocketItem
	{
	public:
		SOCKET mSocket;  
		hlist_node mHlistEntry;
		list_head mListEntry;
		unsigned short mSocketHandle;	
#ifdef WIN32
		HANDLE mhIocp;
		IocpOverLapped mOverLappedRecv;
		IocpOverLapped mOverLappedSend;
#endif
		std::string mRemoteAddress;
		int mRemotePort;
		unsigned long mActiveTick;
	  
		ClientSocketItem()
		{
	    }

		virtual ~ClientSocketItem()
		{
			Close();
			ClearSendBuff();
		}
	  
		void ClearSendBuff()
		{
			SendQueueNode *pTopNode  = NULL;
			mSendBufferMutex.Lock();
			pTopNode = mTopNode;
			mTopNode = NULL;
			mBotNode = NULL;
			mBoSending = false;	
			mSendBufferMutex.UnLock();
#if 1
			SendQueueNode *pdelnode = NULL;
			while(pTopNode){
				pdelnode = pTopNode;
				pTopNode = pTopNode->mNext;
				free(pdelnode->mBuff);
				free(pdelnode);
			}
#endif 
		}

		SOCKET Socket()
		{
			return mSocket;
		}

		void setSocket(SOCKET sock)
		{
			mSocket = sock;
		}

		void ForceClose()
		{
			if(mSocket != INVALID_SOCKET){
				closesocket(mSocket);
				mSocket = INVALID_SOCKET;
				PostQueuedCompletionStatus(mhIocp, 0, (ULONG_PTR)this, (LPOVERLAPPED)DISCONNECT_FLAG);
			}
		}

		virtual void Initialize()
		{
			mSocket = INVALID_SOCKET;
			mRemoteAddress = "";
			mRemotePort = 0;
			mActiveTick = 0;
			mTopNode = NULL;
			mBotNode = NULL;
			mSocketHandle = 0;
			mBoSending = false;
			INIT_HLIST_NODE(&mHlistEntry);
			INIT_LIST_HEAD(&mListEntry);
#ifdef WIN32
			memset(&mOverLappedRecv, 0, sizeof(mOverLappedRecv));
			memset(&mOverLappedSend, 0, sizeof(mOverLappedSend));
			mOverLappedRecv.mOpCode = IOCP_RECV;
			mOverLappedSend.mOpCode = IOCP_SEND;
#endif
		}

		virtual void Finalize()
		{
			mBoSending = false;
			ClearSendBuff();
		}

		void Close()
		{
			ForceClose();
			OnDisConnect();
		}

		virtual void OnError(int) {}
		virtual void OnConnect() {}		
		virtual void OnDisConnect() {}
		virtual void OnSocketRead(char*, unsigned int){}

		void SendBuff(const char* buff, unsigned int bufflen)
		{	
			if(mSocket == INVALID_SOCKET){
				return;
			}
			if (bufflen > 0){
				SendQueueNode* p = (SendQueueNode*)malloc(sizeof(SendQueueNode));
				memset(p, 0, sizeof(SendQueueNode));
				p->mBuff = (char*)malloc(bufflen);
				memmove(p->mBuff, buff, bufflen);
				p->mBuffLen = bufflen;							
				mSendBufferMutex.Lock();
				if (mBotNode){
					mBotNode->mNext = p;
					mBotNode = p;
				}else{
					mTopNode = p;
					mBotNode = p;
				}					
				if(!mBoSending){
					PrepareSend(0, false);
				}
				mSendBufferMutex.UnLock();
			}
		}

#ifdef WIN32
		int PrepareRecv(unsigned int completeBytes)
#else
		int PrepareRecv()
#endif
		{
			if(mSocket == INVALID_SOCKET){
				return 0;
			}
			int result = 0;
#ifdef WIN32	
			OnSocketRead(mOverLappedRecv.mWsaBuf.buf, completeBytes);
			if(mSocket != INVALID_SOCKET){
				PostRecv();
			}
#else
			char recvbuf[MAX_RECV_BUFFERSIZE];
			while(true){
				int ret = ::recv(mSocket, recvbuf, MAX_RECV_BUFFERSIZE, 0);
				if(ret < 0){
					if(errno != EAGAIN){
						OnDisConnect();
						OnError(errno);
						return -1;
					}
					break;
				}else if(ret == 0){
					OnDisConnect();
					return -1;
				}else{
					result += ret;
					OnSocketRead(recvbuf, ret);
					if(ret < MAX_RECV_BUFFERSIZE)
						break;
				}
			}
#endif
			return result;
		}

#ifdef WIN32
		void DoClientSend(int completedBytes)
		{
			int sendlen = mOverLappedSend.mWsaBuf.len - completedBytes;
			if(sendlen > 0){
				memmove(mOverLappedSend.mBuffer,
						mOverLappedSend.mBuffer + completedBytes,
						sendlen);
			}else{
				sendlen = 0;
			}
			PrepareSend(sendlen, true);
		}
#endif

#ifdef WIN32
		void PrepareSend(int sendlen, bool needlock)
#else
		void PrepareSend()
#endif
		{		
#ifdef WIN32
			if(needlock){
				mSendBufferMutex.Lock();
			}
			mBoSending = false;
			while(mTopNode){
				SendQueueNode *node = mTopNode;
				int remainLen = MAX_IOCP_BUF_SIZE - sendlen;
				int datalen = node->mBuffLen - node->mSentBytes;
				if(datalen > remainLen){ 
					memmove(mOverLappedSend.mBuffer + sendlen,
							node->mBuff + node->mSentBytes, 
							remainLen);
					sendlen = MAX_IOCP_BUF_SIZE;
					node->mSentBytes += remainLen;
					break;
				}else{
					memmove(mOverLappedSend.mBuffer + sendlen,
							node->mBuff + node->mSentBytes, 
							datalen);
					sendlen += datalen;
					mTopNode = mTopNode->mNext;
					if(!mTopNode){
						mBotNode = NULL;
					}					
					free(node->mBuff);
					free(node);
				}
			}
			if(sendlen > 0){
				mBoSending = true;
				DWORD dwFlags = 0;
				DWORD dwNumberOfBytesSent = 0;
				mOverLappedSend.mSocketHandle = mSocketHandle;
				mOverLappedSend.mWsaBuf.buf = mOverLappedSend.mBuffer;
				mOverLappedSend.mWsaBuf.len = sendlen;		
				int ret = WSASend(mSocket, &mOverLappedSend.mWsaBuf, 1, &dwNumberOfBytesSent, dwFlags, &mOverLappedSend.mOverLapped, NULL);
				if(ret == SOCKET_ERROR){
					int errCode = WSAGetLastError();
					if(errCode != ERROR_IO_PENDING){
						if(mSocket != INVALID_SOCKET){
							Close();
						}
					}
				}
			}
			if(needlock){
				mSendBufferMutex.UnLock();		
			}
#endif
#ifdef __linux__
			if (!mWorkNode){
				mSendBufferMutex.Lock();
				mWorkNode = mTopNode;
				mTopNode = NULL;
				mBotNode = NULL;
				mBoSending = mWorkNode != NULL;	
				mSendBufferMutex.UnLock();
			}			
			if (mWorkNode){
				AsyncSend();
			}
#endif
		}

#ifdef WIN32
		void PostRecv()
		{
			DWORD dwFlags = 0;
			DWORD dwThancferred = 0;
			mOverLappedRecv.mSocketHandle = mSocketHandle;
			mOverLappedRecv.mWsaBuf.buf = mOverLappedRecv.mBuffer;
			mOverLappedRecv.mWsaBuf.len = MAX_IOCP_BUF_SIZE;
			memset(&mOverLappedRecv.mOverLapped, 0, sizeof(mOverLappedRecv.mOverLapped));
			int ret = WSARecv(mSocket, &mOverLappedRecv.mWsaBuf, 1, &dwThancferred, &dwFlags, &mOverLappedRecv.mOverLapped, NULL);
			if(ret == SOCKET_ERROR){
				int errCode = WSAGetLastError();
				if(errCode != ERROR_IO_PENDING){
					if(mSocket != INVALID_SOCKET){
						Close();
					}
				}
			}
		}
#endif

#ifdef WIN32
		void AsyncSend(SendQueueNode* p)
#else
		void AsyncSend()
#endif 
		{
#ifdef WIN32
			DWORD dwFlags = 0;
			DWORD dwNumberOfBytesSent = 0;
			mOverLappedSend.mSocketHandle = mSocketHandle;
			mOverLappedSend.mWsaBuf.buf = p->mBuff + p->mSentBytes;
			mOverLappedSend.mWsaBuf.len = p->mBuffLen - p->mSentBytes;		
			int ret = WSASend(mSocket, &mOverLappedSend.mWsaBuf, 1, &dwNumberOfBytesSent, dwFlags, &mOverLappedSend.mOverLapped, NULL);
			if(ret == SOCKET_ERROR){
				int errCode = WSAGetLastError();
				if(errCode != ERROR_IO_PENDING){
#if 0
					printf("WSASend error %d\n", errCode);
#endif
					if(mSocket != INVALID_SOCKET){
						Close();
					}
				}
			}
#endif
#ifdef __linux__
			while(mWorkNode){
				int sentbytes = ::send(mSocket, mWorkNode->mBuff + mWorkNode->mSentBytes, mWorkNode->mBuffLen - mWorkNode->mSentBytes, 0);
				if(sentbytes <= 0){
					break;
				}
				mWorkNode->mSentBytes += sentbytes;
				if(mWorkNode->mSentBytes >= mWorkNode->mBuffLen){
					SendQueueNode *pdelnode = mWorkNode;
					mWorkNode = mWorkNode->mNext;
					free(pdelnode->mBuff);
					free(pdelnode);
				}
			}
			mBoSending = mWorkNode != NULL;
#endif
		}

	private:
		bool mBoSending;
		SendQueueNode *mTopNode;
		SendQueueNode *mBotNode;
		a8::sync::Mutex mSendBufferMutex;	
	};

	class ClientSocketPool
	{
		struct ClientSocketNode
		{
			ClientSocketItem* clientsocket;
			unsigned long addtick;
			ClientSocketNode* next;
		};
	public:
		ClientSocketPool():topnode(NULL),
						   botnode(NULL),
						   _count(0)
		{
		}

		~ClientSocketPool()
		{
			ClientSocketNode *pdelnode = topnode;
			while(topnode){
				pdelnode = topnode;
				topnode = topnode->next;
				delete pdelnode->clientsocket;
				free(pdelnode);
			}
		}

		void add(ClientSocketItem* clientsocket)
		{
			ClientSocketNode *p = (ClientSocketNode*)malloc(sizeof(ClientSocketNode));
			memset(p, 0, sizeof(ClientSocketNode));
			p->clientsocket = clientsocket;
			p->addtick = XGetTickCount();
			if(botnode){
				botnode->next = p;
				botnode = p;
			}else{
				topnode = p;
				botnode = p;
			}
			_count++;
		}

		ClientSocketItem* get()
		{
			ClientSocketItem* p = NULL;
			if(topnode){
				if(XGetTickCount() - topnode->addtick >= 30 * 1000){
					p = topnode->clientsocket;
					ClientSocketNode* pdelnode = topnode;
					topnode = topnode->next;
					if(!topnode){
						botnode = NULL;
					}
					free(pdelnode);
					_count--;
				}
			}
			return p;
		}

		void clearTimeOutSocket()
		{
		    unsigned long tick = XGetTickCount();
			while(topnode){
				if(tick - topnode->addtick > 1000 * 60 * 5){
					ClientSocketNode* pdelnode = topnode;
					topnode = topnode->next;
					if(!topnode){
						botnode = NULL;
					}
					delete pdelnode->clientsocket;
					free(pdelnode);
					_count--;
				}else{
					break;
				}
			}
		}

		int count()
		{
			return _count;
		}

	 private:
		ClientSocketNode* topnode;
		ClientSocketNode* botnode;
		int _count;
	};
	
	class TcpListener
	{
	public:
		a8::Delegate<void (a8::ClientSocketItem**)> OnCreateClientSocket;
		a8::Delegate<void(const char*, int, bool&)> OnClientConnect;
		a8::Delegate<void (a8::TcpListener*, int errorId)> OnError;
		std::string mBindAddress;
		unsigned short mBindPort;
	  
		TcpListener(unsigned short maxClientCnt=0xFFFF):
			mListenSocket(INVALID_SOCKET), mBindAddress(""), mBindPort(0),
			mAcceptThread(NULL), mWorkerThread(NULL), mCurrSocketHandle(1000),
			mClientCount(0), mMaxClients(maxClientCnt)
		{
#ifdef WIN32
			mhIocp = 0;
#endif
#ifdef __linux__
			mhEpoll = 0;
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			mhKqueue = 0;
#endif
			INIT_LIST_HEAD(&mClientList);
			INIT_HLIST_BUCKET(mClientsHandleBucket);
		}
		
		virtual ~TcpListener()
		{
			Close();
		}
	 
		void Open()
		{
			if(!isActive())
				setActive(true);
		}
	 
		void Close()
		{
			if(isActive())
				setActive(false);
		}
	 
		bool isActive() 
		{
			return mListenSocket != INVALID_SOCKET;
		}

		void Lock()
		{
			mClientsMutex.Lock();
		}

		void UnLock()
		{
			mClientsMutex.UnLock();
		}

		//prev: Lock posex: Unlock
		bool SocketExists(unsigned short sockhandle)
		{
			return GetClientSocket(sockhandle) != NULL;
		}

		void FreeClient(ClientSocketItem *p)
		{
			mClientsMutex.Lock();
			std::map<ClientSocketItem*, ClientSocketItem*>::iterator itr = mClients.find(p);
			if(itr != mClients.end()){
				mClients.erase(itr);
				if(p->mSocket != INVALID_SOCKET){
					p->ForceClose();
				}
				p->Finalize();
				p->OnDisConnect();
				list_del(&p->mListEntry);
				hlist_del(&p->mHlistEntry);
				mFreeClientPool.add(p);
				mClientCount--;
			}
			mClientsMutex.UnLock();
		}

		void ForceCloseClient(unsigned short sockhandle)
		{
			mClientsMutex.Lock();
			ClientSocketItem *p = GetClientSocket(sockhandle);
			if(p){
				if(p->mSocket != INVALID_SOCKET){
					p->ForceClose();
				}
			}
			mClientsMutex.UnLock();
		}

		bool SendClientMsg(unsigned short sockhandle, const char *buff, int buffLen)
		{
			bool boSendOk = false;
			mClientsMutex.Lock(); 
			ClientSocketItem *p = GetClientSocket(sockhandle);
			if (p){
				p->SendBuff(buff, buffLen);
				boSendOk = true;
			}		 
			mClientsMutex.UnLock();
			return boSendOk;
		}

		bool SendClientMsg(unsigned short sockhandle, const std::string& buff)
		{
			return SendClientMsg(sockhandle, buff.data(), buff.size());
		}

		unsigned int GetClientSocketCount()
		{
			return mClientCount;
		}

		void ForeachSocket(a8::Delegate<void (a8::ClientSocketItem*, std::vector<unsigned short>&) >& func,
				std::vector<unsigned short>& vec)
		{
			Lock();
			struct list_head *node;
			list_for_each(node, &mClientList){
				ClientSocketItem *p = list_entry(node, ClientSocketItem, mListEntry);
				func(p, vec);
			}
			UnLock();
		}

	private:
		SOCKET mListenSocket;
		a8::Runnable* mAcceptThread;
		a8::Runnable* mWorkerThread;
		a8::sync::Mutex mClientsMutex;	
		hlist_head mClientsHandleBucket[1024];
		std::map<ClientSocketItem*, ClientSocketItem*> mClients;
		list_head mClientList;
		volatile unsigned int mClientCount;
		unsigned short mCurrSocketHandle;
		unsigned short mMaxClients;
		ClientSocketPool mFreeClientPool;
#ifdef __linux__
		int mhEpoll;
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
		int mhKqueue;
#endif
#ifdef WIN32
		HANDLE mhIocp;
#endif
	  
		void setActive(bool active) 
		{ 
			if(active){
				if(isActive()) 
					return;
				if(!mWorkerThread){
					mWorkerThread = new a8::Runnable();
					mWorkerThread->OnExecute += std::make_pair(this, &TcpListener::WorkerThreadProc);
					mWorkerThread->Start();
				}
			}else{
				if(isActive())
					ActiveStop();		
			}
		}

		bool ActiveStart()
		{
#ifdef WIN32
			mListenSocket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
#else
			mListenSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
#endif
			if(INVALID_SOCKET == mListenSocket){
				OnError(this, errno);
				return false;
			}
			// TIME_WAIT - argh
			int on = 1;
			if (setsockopt(mListenSocket, SOL_SOCKET, SO_REUSEADDR, (const char*)&on, sizeof(on)) == -1 ){
				//return false;
			}
			sockaddr_in sa;
			memset(&sa, 0, sizeof(sa));
			sa.sin_family = AF_INET;
			sa.sin_addr.s_addr = inet_addr(mBindAddress.c_str());
			sa.sin_port = htons(mBindPort);
			if(bind(mListenSocket, (sockaddr*)&sa, sizeof(sa)) < 0){
				OnError(this, errno);
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;
				return false;
			}
			if(listen(mListenSocket, mMaxClients) < 0) {
				OnError(this, errno);
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;
				return false;
			}
#ifdef __linux__
			mhEpoll = epoll_create(mMaxClients);
			assert(mhEpoll != -1);
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			mhKqueue = kqueue();
#endif
#ifdef WIN32
			mhIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);
			assert(mhIocp);
#endif		
			mAcceptThread = new a8::Runnable();
			mAcceptThread->OnExecute += std::make_pair(this, &TcpListener::AcceptThreadProc);
			mAcceptThread->Start();
			return true;
		}

		void ActiveStop()
		{
#ifdef WIN32
			if (mhIocp){
				PostQueuedCompletionStatus(mhIocp, 0, NULL, NULL);
			}
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			if(mhKqueue){
				close(mhKqueue);
				mhKqueue = 0;
			}
#endif
			if(mListenSocket != INVALID_SOCKET){
				closesocket(mListenSocket);
				mListenSocket = INVALID_SOCKET;								
			}
			if (mAcceptThread){
				delete mAcceptThread;
				mAcceptThread = NULL;
			}		
			if (mWorkerThread){
				delete mWorkerThread;
				mWorkerThread = NULL;
			}
#ifdef __linux__
			if(mhEpoll != -1){
				close(mhEpoll);
				mhEpoll = -1;
			}
#endif
#ifdef WIN32
			if (mhIocp){
				CloseHandle(mhIocp);
				mhIocp = NULL;
			}
#endif		
		}
	
		void AcceptThreadProc(a8::Runnable* thread)
		{
			socklen_t addrLen = 0;
			sockaddr_in addr;
			while(!thread->IsShutdown()){
				if(INVALID_SOCKET == mListenSocket)
					break;	
				addrLen = sizeof(sockaddr_in);
				memset(&addr, 0, sizeof(addr));
				SOCKET sock = accept(mListenSocket, (sockaddr*)&addr, &addrLen);
				if(INVALID_SOCKET != sock){
					if (thread->IsShutdown()){
						closesocket(sock);
					}else
						if(!SocketAccept(sock, inet_ntoa(addr.sin_addr), addr.sin_port))
							closesocket(sock);
				}
			}
		}

		void WorkerThreadProc(a8::Runnable* thread)
		{
			if(!ActiveStart())
				return;
#ifdef WIN32
			while (!thread->IsShutdown()){
				DWORD completeBytes = 0;
				OVERLAPPED *ol = NULL;
				ClientSocketItem *clientsocket = NULL;
				BOOL boSuccess = GetQueuedCompletionStatus(mhIocp, &completeBytes,(PULONG_PTR)&clientsocket,(LPOVERLAPPED*)&ol, INFINITE);
				if(ol == (LPOVERLAPPED)DISCONNECT_FLAG && clientsocket){
					FreeClient(clientsocket);
					continue;		
				}
				// I/O¶ÁÐ´·µ»Ø0£¬¿ÉÄÜ³¬Ê±£¬¿ÉÄÜ¿Í»§¶Ë¶Ï¿ªÁ¬½Ó¡£  µÚÒ»¸ö·´Ó¦
				if(!boSuccess || completeBytes == 0){
					if(clientsocket){
						clientsocket->ForceClose();
					}
					continue;
				}
				if (clientsocket && ol){
					IocpOverLapped* iocpOl = container_of(ol, IocpOverLapped, mOverLapped);
					clientsocket->mActiveTick = 0;
					switch(iocpOl->mOpCode){
					case IOCP_RECV:		
						{
							clientsocket->PrepareRecv(completeBytes);
						}
						break;
					case IOCP_SEND:
						{
							clientsocket->DoClientSend(completeBytes);
						}
						break;
					}
				}
			}	 	  
#endif
#ifdef __linux__
			epoll_event *events = new epoll_event[mMaxClients];
			while(!thread->IsShutdown()){
				int nfds = epoll_wait(mhEpoll, events, mMaxClients, -1);
				for(int i = 0; i < nfds; i++){	  
					if(events[i].events & EPOLLIN){
						ClientSocketItem *clientsocket = (ClientSocketItem*)events[i].data.ptr;
						if(clientsocket->PrepareRecv() < 0){
							mClientsMutex.Lock();
							hlist_del(&clientsocket->mHlistEntry);
							list_del(&clientsocket->mListEntry);
							mClientsMutex.UnLock();
							struct epoll_event ev;
							epoll_ctl(mhEpoll, EPOLL_CTL_DEL, clientsocket->Socket(), &ev);
							clientsocket->Finalize();
							delete clientsocket;
						}
					}else if(events[i].events & EPOLLOUT){
						ClientSocketItem *clientsocket = (ClientSocketItem*)events[i].data.ptr;
						clientsocket->PrepareSend();
					}
				}
			}
			delete [] events;
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			struct kevent *events = (struct kevent*)malloc(sizeof(struct kevent) * mMaxClients);
			while(!thread->IsShutdown()){
				int nEvents = kevent(mhKqueue, NULL, 0, events, mMaxClients, NULL);
				if(nEvents == -1)
					break;
				for(int i = 0; i < nEvents; i++){
				ClientSocketItem* clientSocket = (ClientSocketItem*) events[i].udata;
					if(events[i].flags & EV_EOF){
						clientSocket->OnDisConnect(clientSocket);
						clientSocket->Finalize();
						mClientsMutex.Lock(); 
						hlist_del(&clientSocket->mHlistEntry);
						list_del(&clientSocket->mListEntry);
						mClientsMutex.UnLock();
						delete clientSocket;
						continue; 
					}
					if(events[i].filter == EVFILT_READ )
						clientSocket->PrepareRecv();
					if(events[i].filter == EVFILT_WRITE )
						clientSocket->PrepareSend(0);
				}
			}
			free(events);
#endif
		}

		bool SocketAccept(SOCKET sock, const char* ipAddr, int port)
		{		
			if(mClientCount >= mMaxClients)	
				return false;
			bool refuse = false;
			OnClientConnect(ipAddr, port, refuse);
			if (refuse){
				return false;
			}

			mClientsMutex.Lock();
			mFreeClientPool.clearTimeOutSocket();
			ClientSocketItem* p = mFreeClientPool.get();
			mClientsMutex.UnLock();

			if(!p){
				OnCreateClientSocket(&p);
			}
			if (!p){
				return false;
			}
			p->Initialize();
			p->setSocket(sock);
			p->mRemoteAddress = std::string(ipAddr);
			p->mRemotePort = port;

			mClientsMutex.Lock();
			++mCurrSocketHandle;			
			while (GetClientSocket(mCurrSocketHandle)){
				if (mCurrSocketHandle < 1000){
					mCurrSocketHandle = 1000;
				}
				++mCurrSocketHandle;
			}	
			p->mSocketHandle = mCurrSocketHandle;			
			list_add(&p->mListEntry, &mClientList);		
			int h = p->mSocketHandle % a8::arraysize(mClientsHandleBucket);			
			hlist_add_head(&p->mHlistEntry, &mClientsHandleBucket[h]);
			mClientCount++;			
			mClients.insert(std::make_pair(p, p));
			assert(GetClientSocket(p->mSocketHandle));
			mClientsMutex.UnLock();
			p->OnConnect();
#ifdef WIN32
			p->mhIocp = mhIocp;
			if(CreateIoCompletionPort((HANDLE)sock, mhIocp, (ULONG_PTR)p, 0) == 0){
				printf("CreateIoCompletionPort error %d\n", ::GetLastError());
				p->ForceClose();
			}else{
				p->PostRecv();
			}
#endif
#ifdef __linux__
			struct epoll_event ev;
			ev.data.fd = sock;
			ev.events = EPOLLIN | EPOLLOUT | EPOLLET;
			ev.data.ptr = p;
			int n = epoll_ctl(mhEpoll, EPOLL_CTL_ADD, sock, &ev);
			assert(n == 0);
#endif
#if defined(__FREEBSD__) || defined(__APPLE__)
			struct kevent kvt;
			memset(&kvt, 0, sizeof(struct kevent));
			EV_SET(&kvt, sock, EVFILT_READ, EV_ADD | EV_ENABLE | EV_ERROR, 0, 0, p);
			kevent(mhKqueue, &kvt, 1, NULL, 0, NULL);
			memset(&kvt, 0, sizeof(struct kevent));
			EV_SET(&kvt, sock, EVFILT_WRITE, EV_ADD | EV_ENABLE | EV_ERROR, 0, 0, p);
			kevent(mhKqueue, &kvt, 1, NULL, 0, NULL);
#endif
			return true;
		}
	  
		ClientSocketItem* GetClientSocket(unsigned short handle)
		{
			int h = handle % a8::arraysize(mClientsHandleBucket);
			hlist_head *hlist = &mClientsHandleBucket[h];
			hlist_node *node = NULL;
			hlist_for_each(node, hlist){
				ClientSocketItem* p = hlist_entry(node, ClientSocketItem, mHlistEntry);
				if (p->mSocketHandle == handle)
					return p;			
			}
			return NULL;
		}
	};
}

#endif
